package com.zyx.flinkdemo.cdc.stream;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @author Yaxi.Zhang
 * @since 2021/6/24 11:01
 * desc: FlinkCDC入门案例
 */
public class StreamCdcDemo {
    public static void main(String[] args) throws Exception {
        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //开启CK
        // env.enableCheckpointing(5000L);
        // env.getCheckpointConfig().setCheckpointTimeout(5000L);
        // env.setStateBackend(new HashMapStateBackend());
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://bigdata021:8020/dev/cdctest");


        //2.使用CDC的方式读取MySQL变化数据
        DebeziumSourceFunction<String> sourceFunction = MySQLSource
                .<String>builder()
                .hostname("10.1.207.12")
                .port(3306)
                .username("root")
                .password("root")
                .serverTimeZone("Asia/Shanghai")
                .databaseList("db_report_amis")
                .tableList("db_report_amis.test")
                .startupOptions(StartupOptions.initial())
                // .startupOptions(StartupOptions.earliest())
                // .startupOptions(StartupOptions.latest())
                // .startupOptions(StartupOptions.specificOffset())
                // .startupOptions(StartupOptions.timestamp())
                .deserializer(new StringDebeziumDeserializationSchema())
                .build();
        DataStreamSource<String> streamSource = env.addSource(sourceFunction);

        //3.打印
        streamSource.print();

        //4.启动
        env.execute();
    }
}
